
#region Imports

using System;
using System.Collections.Generic;
using System.Messaging;
using System.Text;
using System.Xml;

using Justa.Common;

#endregion

namespace Justa.Messaging
{
    /// <summary>
    /// 
    /// </summary>
    /// 
    /// <author>Abel Perez (java.aperez@gmail.com)</author>
    /// 
    public class MessageGateway : GenericMessageGateway, IMessageGateway
    {
        /// <summary>
        /// 
        /// </summary>
        /// 
        public event MessageEventHandler OnMessageRecieved;

        /// <summary>
        /// 
        /// </summary>
        private MessageQueue destination;

        /// <summary>
        /// 
        /// </summary>
        /// 
        private List<MessageQueue> queues;

        /// <summary>
        /// 
        /// </summary>
        /// 
        private IDictionary<string, MessageEventHandler> registry;

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        public MessageGateway(String destination)
        {
            this.destination = GetMessageQueue(destination);

            if (destination == null)
            {
                throw new ApplicationException("Failed to initialize MessageGateway.");
            }

             queues = new List<MessageQueue>();
             registry = new Dictionary<string, MessageEventHandler>();
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="machine"></param>
        /// <param name="threshold"></param>
        /// 
        /// <returns></returns>
        /// 
        public List<QueueSummary> GetSuspectQueues(string machine, int threshold)
        {
            List<QueueSummary> result = new List<QueueSummary>();
            MessageQueue[] messageQueues = MessageQueue.GetPrivateQueuesByMachine(machine);

            if (messageQueues.Length > 0)
            {
                foreach (MessageQueue queue in messageQueues)
                {
                    if (queue.GetAllMessages().Length > threshold)
                    {
                        QueueSummary queueSummary = new QueueSummary();
                        result.Add(new QueueSummary(queue.QueueName, queue.GetAllMessages().Length));
                    }
                }
            }

            return result;
        }
        
        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="message"></param>
        /// 
        public void Send(string message)
        {
            ArgumentValidator.ValidateString(message, "message");
            OneWay(message);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="message"></param>
        /// 
        public void Send(XmlDocument message)
        {
            ArgumentValidator.Validate(message, "message");
            Send(message.OuterXml);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        /// <param name="message"></param>
        /// 
        public void Send(string destination, Message message)
        {
            ArgumentValidator.ValidateString(destination, "destination");
            ArgumentValidator.Validate(message, "message");

            OneWay(message);
        }
        
        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        /// <param name="message"></param>
        /// 
        public void Send(string destination, string message)
        {
            ArgumentValidator.ValidateString(destination, "destination");
            ArgumentValidator.ValidateString(message, "message");
            OneWay(GetMessageQueue(destination), message);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        /// <param name="message"></param>
        /// 
        public void Send(string destination, XmlDocument message)
        {
            ArgumentValidator.Validate(message, "message");
            Send(destination, message.OuterXml);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="responseDestination"></param>
        /// <param name="message"></param>
        /// 
        /// <returns></returns>
        /// 
        public string Request(string responseDestination, string message)
        {
            return RequestResponse(destination, responseDestination, message, 0);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="message"></param>
        /// <returns></returns>
        /// 
        public string Request(string responseDestination, XmlDocument message)
        {
            ArgumentValidator.Validate(message, "message");
            return Request(responseDestination, message.OuterXml);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="responseDestination"></param>
        /// <param name="message"></param>
        /// <param name="timeout"></param>
        /// 
        /// <returns></returns>
        /// 
        public string Request(string responseDestination, string message, int timeout)
        {
            return RequestResponse(destination, responseDestination, message, timeout);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="responseDestination"></param>
        /// <param name="message"></param>
        /// <param name="timeout"></param>
        /// 
        /// <returns></returns>
        /// 
        public string Request(string responseDestination, XmlDocument message, int timeout)
        {
            ArgumentValidator.Validate(message, "message");
            return Request(responseDestination, message.OuterXml, timeout);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="targetDestination"></param>
        /// <param name="responseDestination"></param>
        /// <param name="message"></param>
        /// <param name="timeout"></param>
        /// 
        /// <returns></returns>
        /// 
        public string Request(string targetDestination, string responseDestination, string message, int timeout)
        {
            ArgumentValidator.ValidateString(targetDestination, "targetDestination");
            return RequestResponse(GetMessageQueue(targetDestination), responseDestination, message, timeout);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="targetDestination"></param>
        /// <param name="responseDestination"></param>
        /// <param name="message"></param>
        /// <param name="timeout"></param>
        /// 
        /// <returns></returns>
        /// 
        public string Request(string targetDestination, string responseDestination, XmlDocument message, int timeout)
        {
            ArgumentValidator.Validate(message, "message");
            return Request(targetDestination, responseDestination, message.OuterXml, timeout);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="message"></param>
        private void OneWay(Object message)
        {
            try
            {
                destination.Send(message);
            }
            catch (MessageQueueException exception)
            {
                throw new ApplicationException("Failed to send message.", exception);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="messageQueue"></param>
        /// <param name="message"></param>
        /// 
        private void OneWay(MessageQueue messageQueue, Object message)
        {
            try
            {
                messageQueue.Send(message);
            }
            catch (MessageQueueException exception)
            {
                throw new ApplicationException("Failed to send message.", exception);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="targetDestination"></param>
        /// <param name="responseDestination"></param>
        /// <param name="msg"></param>
        /// <param name="timeout"></param>
        /// 
        /// <returns></returns>
        /// 
        private string RequestResponse(MessageQueue targetDestination, string responseDestination, string msg, int timeout)
        {
            targetDestination.MessageReadPropertyFilter.CorrelationId = true;
            MessageQueue responseQueue = GetMessageQueue(responseDestination);
            responseQueue.MessageReadPropertyFilter.CorrelationId = true;

            Message message = new Message(msg);
            message.ResponseQueue = responseQueue;
           
            OneWay(targetDestination, message);
            string correlationId = message.Id;
            //string correlationId = message.CorrelationId;
            Console.WriteLine("correlation id: {0}", correlationId);

            Message response = null;
            if (timeout <= 0)
            {
                //response = responseQueue.ReceiveById(correlationId);
                response = responseQueue.ReceiveByCorrelationId(correlationId);
            }
            else
            {
                //response = responseQueue.ReceiveById(correlationId, new System.TimeSpan(0, 0, timeout));
                response = responseQueue.ReceiveByCorrelationId(correlationId, new System.TimeSpan(0, 0, timeout));
            }

            response.Formatter = new System.Messaging.XmlMessageFormatter(new String[] { "System.String" });
            return response.Body.ToString();
        }
       
        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        /// <param name="messageEventHandler"></param>
        /// 
        public void Receive(string destination, MessageEventHandler messageEventHandler)
        {
            ArgumentValidator.Validate(messageEventHandler, "messageEventHandler");
            
            string targetDestination = GetFullyQualifiedQueueName(destination);
            Console.WriteLine("listening on: {0}", GetFullyQualifiedQueueName(destination));

            if (!registry.ContainsKey(targetDestination))
            {
                lock (registry)
                {
                    registry.Add(targetDestination, messageEventHandler);
                }
            }
            else
            {
                registry[targetDestination] += messageEventHandler;
            }

            Start(destination);
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="destination"></param>
        /// 
        public void StopReceiving(string destination)
        {
            ArgumentValidator.ValidateString(destination, "destination");

            string targetDestination = GetFullyQualifiedQueueName(destination);

            for (int i = 0; i < queues.Count; i++)
            {
                Console.WriteLine("searching for queues to remove...");
                if (targetDestination.Equals(queues[i].QueueName))
                {
                    Console.WriteLine("found one.");
                    queues[i].Close();
                    queues[i] = null; //.RemoveAt(i);
                    break;
                 }
            }
        }

        /// <summary>
        /// Starts the asynchronous recieve process for this gateway.
        /// </summary>
        /// 
        public void Start(string destination)
        {
            try
            {
                MessageQueue messageQueue = GetMessageQueue(destination);
                messageQueue.Formatter = new System.Messaging.XmlMessageFormatter(new String[] { "System.String" });
                
                messageQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(OnReceiveCompleted);
                messageQueue.BeginReceive();
                queues.Add(messageQueue);
                
            }
            catch (MessageQueueException exception)
            {
                throw new ApplicationException("Failed to initialize MessageReceiverGateway.", exception);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// 
        /// <param name="source"></param>
        /// <param name="asyncResult"></param>
        /// 
        private void OnReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
        {
            MessageQueue destination = null;

            try
            {
                destination = (MessageQueue)source;
                Message message = destination.EndReceive(asyncResult.AsyncResult);
                Console.WriteLine("received message event from {0}", destination.QueueName);
                registry[destination.QueueName](this, new MessageEventArgs(message));
            }
            catch (Exception exception)
            {
                Console.WriteLine(exception.Message);
            }
            finally
            {
                // resume async receive
                destination.BeginReceive();
            }
        }

        /// <summary>
        /// Default operation that will be invoked for each incomming message.
        /// </summary>
        /// 
        /// <param name="message">The incomming message received by this gateway.</param>
        /// 
        private void DefaultEvent(object sender, MessageEventArgs message)
        {
        }
    }
}
